use crate::core::error::operation_error::OperationError;
use crate::core::operation::download::Error::NoResponse;
use crate::core::operation::Operation;
use crate::core::path_type::TypedPath;
use crate::core::step_context::StepContext;
use chrono::{TimeDelta, Utc};
use http::Uri;
use snafu::{ResultExt, Snafu};
use std::sync::Arc;
use tokio::fs::File;
use tokio::io::copy;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use tokio::task::JoinHandle;
use tokio_stream::StreamExt;
use tonic::async_trait;

#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
pub enum Error {
    #[snafu(display("Failed to get response from {url}"))]
    NoResponse { url: String },

    #[snafu(display("An io error occurred: {source}"))]
    IO { source: std::io::Error },

    #[snafu(display("Failed to convert response body to stream of bytes: {source}"))]
    ConvertResponse {
        #[snafu(source(from(reqwest::Error, Box::new)))]
        source: Box<dyn std::error::Error + Send + Sync>,
    },
}

impl From<Error> for OperationError {
    fn from(source: Error) -> Self {
        OperationError::Download { source }
    }
}

pub struct Download {
    uri: Uri,
    dst: TypedPath,
}

const LOG_INTERVAL: u32 = 1;

fn log_progress(
    max: Option<u64>,
    mut cur: Receiver<u64>,
    context: Arc<StepContext>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        if let Some(m) = max {
            while let Some(c) = cur.recv().await {
                let percent = (c as f64) / (m as f64) * 100.0;
                context.log(format!(
                    "Downloaded {} bytes, total {} bytes, completed: {:.2}%",
                    c, m, percent
                ));
            }
        } else {
            while let Some(c) = cur.recv().await {
                context.log(format!("Downloaded {} bytes", c));
            }
        }
    })
}

#[async_trait]
impl Operation for Download {
    async fn run(&mut self, context: Arc<StepContext>) -> Result<(), OperationError> {
        let rsp = reqwest::get(self.uri.to_string())
            .await
            .map_err(|e| NoResponse {
                url: self.uri.to_string(),
            })?;

        let dst = match &self.dst {
            TypedPath::Normal(p) => p.clone(),
            TypedPath::Contextual(s) => context.job_context().path().join(s),
        };

        let mut dest = File::create(dst).await.context(IOSnafu {})?;

        let max = rsp.content_length();
        let mut content = rsp.bytes_stream();
        let (sender, receiver) = mpsc::channel(8);
        let progress = log_progress(max, receiver, context.clone());

        let mut count = 0;
        let mut log_time = Utc::now();
        while let Some(chunk) = content.next().await {
            let chunk = chunk.context(ConvertResponseSnafu {})?;
            let l: u64 = chunk.len().try_into().unwrap();
            count += l;
            let now = Utc::now();
            if now > log_time + TimeDelta::seconds(1) {
                sender.send(count).await.unwrap();
                log_time = now;
            }
            sender.send(count).await.unwrap();
            copy(&mut chunk.as_ref(), &mut dest)
                .await
                .context(IOSnafu {})?;
        }

        progress.abort();

        Ok(())
    }
}

impl Download {
    pub fn dst(&self) -> &TypedPath {
        &self.dst
    }

    pub fn new(uri: Uri, dst: TypedPath) -> Download {
        Download { uri, dst }
    }
}
